package com.example.flinktest.transformation;

import com.example.flinktest.source.Event;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

public class TransTupleAggregationTest {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        DataStreamSource<Tuple2<String, Integer>> streamSource = env.fromElements(
                Tuple2.of("a", 1),
                Tuple2.of("a", 2),
                Tuple2.of("b", 3),
                Tuple2.of("b", 4)
        );
        //数据流是对象
        DataStreamSource<Event> source = env.fromElements(
                new Event("JunSongLi", "./home", 100L),
                new Event("Tomas", "./cart", 120L)
        );
//        streamSource.keyBy(r -> r.f0).sum(1).print();
//        streamSource.keyBy(r -> r.f0).min(1).print();
//        streamSource.keyBy(r -> r.f0).max(1).print();
//        streamSource.keyBy(r -> r.f0).max("f1").print();
//        streamSource.keyBy(r -> r.f0).min(1).print();
//        streamSource.keyBy(r -> r.f0).min("f1").print();
        streamSource.keyBy(r -> r.f0).maxBy(1).print();
//        streamSource.keyBy(r -> r.f0).maxBy("f1").print();
//        streamSource.keyBy(r -> r.f0).minBy(1).print();
//        streamSource.keyBy(r -> r.f0).minBy("f1").print();
        source.keyBy(e -> e.user).max("timestamp").print();
        env.execute();
    }
}
